package com.wk.job.heartbeat.master;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.concurrent.ConcurrentHashMap;

import com.wk.database.DB_Mysql;
import com.wk.database.MachineState;
import com.wk.database.UpdatedSlaveMachine;
import com.wk.database.entity.Update;
import com.wk.job.Job;
import com.wk.job.Task;
import com.wk.job.heartbeat.JobStateHeartbeat;
import com.wk.net.heartbeat.handler.HeartbeatListener;
import com.wk.net.heartbeat.handler.HeartbeatObserver;
import com.wk.net.heartbeat.handler.ParsedHeartbeatMessage;

class JobStateListener implements HeartbeatListener<JobStateHeartbeat> {

	private ConcurrentHashMap<Integer, Task> tasks;
	private ConcurrentHashMap<String, Job> jobs;

	private enum ObservationState {
		OBSERVED, NOT_OBSERVED, DEPRECATED
	}

	public JobStateListener(ConcurrentHashMap<String, Job> jobs, ConcurrentHashMap<Integer, Task> tasks) {
		this.jobs = jobs;
		this.tasks = tasks;

		Connection connection = DB_Mysql.getConnection();
		if (connection == null) {
			System.out.println("[AbsentSlaveListener]连接数据库失败");
			this.updateMachine = null;
			return;
		}

		Update<UpdatedSlaveMachine> update = new Update<>(UpdatedSlaveMachine.class);
		try {
			update.connect(DB_Mysql.getConnection());
		} catch (SQLException e) {
			this.updateMachine = null;
			e.printStackTrace();
		}

		this.updateMachine = update;
	}

	@Override
	public void onHeartbeat(ParsedHeartbeatMessage<JobStateHeartbeat> message) {
		System.out.println("收到心跳，来自：" + message.source + "；已监视？" + message.isSourceObserved);
		System.out.println("内容：" + message.heartbeat.toString());

		updateJob(message);
		updateMachineState(message);
	}

	@Override
	public void alterObservation(ParsedHeartbeatMessage<JobStateHeartbeat> message, HeartbeatObserver observer) {
		synchronized (observer) {
			switch (recentReportingMachineState) {
			case NOT_OBSERVED:
				System.out.println("新增监视：" + message.source);
				observer.observe(message.source, true);
				break;
			case DEPRECATED:
				System.out.println("取消监视：" + message.source);
				observer.ignore(message.source);
				break;
			default:
				// 保持监视，什么都不做
				break;
			}
		}
	}

	private void updateJob(ParsedHeartbeatMessage<JobStateHeartbeat> message) {
		JobStateHeartbeat heartbeat = message.heartbeat;

		if (!heartbeat.isUnderway) {
			return;
		}

		Job job = jobs.get(heartbeat.jobId);
		if (job == null) {
			return;
		}

		job.setJobInfo(heartbeat.progress);
		if (heartbeat.isFailed) {
			job.setFailed();
		}

		if (job.isFinished() || job.isFailed()) {
			System.out.println("Job已结束。成功？" + job.isFinished() + "失败？" + job.isFailed());
			if (tasks != null) {
				updateTask(job);
			}
		} else {
			System.out.println("Job未结束");
		}
	}

	private void updateMachineState(ParsedHeartbeatMessage<JobStateHeartbeat> message) {
		int countUpdated = setMachineOnline(message.source);

		if (!message.isSourceObserved && countUpdated != 0) {
			System.out.println("发现新机器：" + message.source);
			recentReportingMachineState = ObservationState.NOT_OBSERVED;
		} else if (message.isSourceObserved && countUpdated == 0) {
			System.out.println("机器已失效：" + message.source);
			recentReportingMachineState = ObservationState.DEPRECATED;
		} else {
			recentReportingMachineState = ObservationState.OBSERVED;
		}
	}

	private int setMachineOnline(String machine) {
		int countUpdated;
		try {
			countUpdated = updateMachine.execute(new UpdatedSlaveMachine(machine, MachineState.NORMAL));
			System.out.println("更新机器状态为在线。更新数：" + countUpdated);
		} catch (SQLException e) {
			e.printStackTrace();
			countUpdated = 0;
		}
		return countUpdated;
	}

	private void updateTask(Job job) {
		int taskId = job.getTaskId();
		Task task = tasks.get(taskId);
		System.out.println(task.toString());
		task.removeAjob(job);
		jobs.remove(job.getId());

		System.out.println("task：" + taskId + "；剩余Job数：" + task.countRemainingJobs());
		if (task.isFinished()) {
			System.out.println("task结束：" + taskId + "。成功？" + !task.isFailed());
			if (task.isFailed()) {
				DB_Mysql.setFailed(taskId);
			} else {
				DB_Mysql.setFinished(taskId);
			}
		}

		System.out.println("task已完成?:" + task.isFinished() + " ->" + task.toString());
	}

	private Update<UpdatedSlaveMachine> updateMachine;
	private ObservationState recentReportingMachineState;
}
